package com.doeis.minaserver.nio;

import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;

import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 参照 NioEventLoop
 * **/
public class NioServiceEventLoop extends ServiceEventExecutor {

	private static final int ST_NOT_STARTED = 1;
	private static final int ST_STARTED = 2;
	private static final int ST_SHUTTING_DOWN = 3;
	private static final int ST_SHUTDOWN = 4;
	private volatile int state = ST_NOT_STARTED;

	private final Object stateLock = new Object();

	private static final InternalLogger logger = InternalLoggerFactory
			.getInstance(NioServiceEventLoop.class);

	private final Executor executor;

	private volatile Thread thread;

	private final Queue<Runnable> taskQueue;

	protected NioServiceEventLoop(NioServiceEventLoopGroup parent,
			Executor executor) {
		this.executor = executor;
		taskQueue = new LinkedBlockingQueue<Runnable>();
	}

	@Override
	public void execute(Runnable task) {
		if (task == null) {
			throw new NullPointerException("task");
		}

		boolean inEventLoop = inEventLoop();
		if (inEventLoop) {
			addTask(task);
		} else {
			startThread();//开启线程
			addTask(task);
			if (isShutdown() && removeTask(task)) {
				reject();
			}
		}
	}

	private void startThread() {
		synchronized (stateLock) {
			if (state == ST_NOT_STARTED) {
				state = ST_STARTED;
				doStartThread();
			}
		}
	}

	private void doStartThread() {
		assert thread == null;
		executor.execute(new Runnable() {
			@Override
			public void run() {
				thread = Thread.currentThread();

				boolean success = false;
				try {
					thread.setName(thread.getName()+ "NioServiceEventLoop");
					NioServiceEventLoop.this.run();
					success = true;
				} catch (Throwable t) {
					logger.warn(
							"Unexpected exception from an event executor: ", t);
				} finally {
					if (state < ST_SHUTTING_DOWN) {
						state = ST_SHUTTING_DOWN;
					}
					// 需要处理
				}
			}
		});
	}

	public void shutdownGracefully() {
		boolean inEventLoop = inEventLoop();
		synchronized (stateLock) {
			if (isShuttingDown()) {
				return;
			}

			if (inEventLoop) {
				assert state == ST_STARTED;
				state = ST_SHUTTING_DOWN;
			} else {
				switch (state) {
				case ST_NOT_STARTED:
					state = ST_SHUTTING_DOWN;
					doStartThread();
					break;
				case ST_STARTED:
					state = ST_SHUTTING_DOWN;
					break;
				default:
					
				}
			}
		}
	}

	protected void run() {
		for (;;) {
			try{
				runAllTasks();
			}catch(Exception e){
				  try {
					Thread.sleep(1000);
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				}
			}
			
		}
	}

	/**
	 * @see {@link Queue#remove(Object)}
	 */
	protected boolean removeTask(Runnable task) {
		if (task == null) {
			throw new NullPointerException("task");
		}
		return taskQueue.remove(task);
	}

	protected void addTask(Runnable task) {
		if (task == null) {
			throw new NullPointerException("task");
		}
		if (isShutdown()) {
			reject();
		}
		taskQueue.add(task);
	}

	private void reject() {

	}

	protected boolean runAllTasks() {
		Runnable task = takeTask();//在这个等着任务到来，如果没有任务一直卡着
		if (task == null) {
			return false;
		}

		for (;;) {
			try {
				task.run();//执行任务
			} catch (Throwable t) {
				logger.warn("A task raised an exception.", t);
			}

			task = pollTask();//再次获取是否有任务执行，如果有任务则一直执行
			if (task == null) {
				return true;
			}
		}
	}

	private static final Runnable WAKEUP_TASK = new Runnable() {
		@Override
		public void run() {
			// Do nothing.
		}
	};

	/**
	 * Retrieves and removes the head of this queue,
     * or returns <tt>null</tt> if this queue is empty.
	 * @see {@link Queue#poll()}
	 */
	protected Runnable pollTask() {
		assert inEventLoop();
		for (;;) {
			Runnable task = taskQueue.poll();
			if (task == WAKEUP_TASK) {
				continue;
			}
			return task;
		}
	}
	
	/**
     * @see {@link Queue#isEmpty()}
     */
    protected boolean hasTasks() {
        assert inEventLoop();
        return !taskQueue.isEmpty();
    }
    
    
    protected Runnable takeTask() {
        assert inEventLoop();
        if (!(taskQueue instanceof BlockingQueue)) {
            throw new UnsupportedOperationException();
        }
        BlockingQueue<Runnable> taskQueue = (BlockingQueue<Runnable>) this.taskQueue;
        for (;;) {
        	  Runnable task = null;
              try {
                  task = taskQueue.take();
                  //取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到
                  //　BlockingQueue有新的数据被加入;
                  if (task == WAKEUP_TASK) {
                      task = null;
                  }
              } catch (InterruptedException e) {
                  // Ignore
              }
              return task;
        }
    }

	public boolean inEventLoop() {
		return inEventLoop(Thread.currentThread());
	}

	public boolean inEventLoop(Thread thread) {
		return thread == this.thread;
	}

	/**
	 * 这个线程是否已经停止
	 * **/
	public boolean isShutdown() {
		return state >= ST_SHUTDOWN;
	}

	public boolean isShuttingDown() {
		return state >= ST_SHUTTING_DOWN;
	}

}
